"""
Case Type   : DBlink功能
Case Name   : dblink不同session操作相同表
Create At   : 2024/6/15
Owner       : peilinqian
Description :
    step1: 创建数据库及库下创建dblink扩展;
    step2: 目标库进行数据库及用户建立;
    step3: 目标库进行pg_hba用户白名单配置;
    step4: sql脚本目标库信息变更配置;
    step5: 目标库进行数据预置;
    step6: 以线程开启事务，同时进行10个dblink连接数据插入;
Expect      :
    step1: 创建数据库及库下创建dblink扩展; expect:成功
    step2: 目标库进行数据库及用户建立; expect:成功
    step3: 目标库进行pg_hba用户白名单配置; expect:成功
    step4: sql脚本目标库信息变更配置; expect:成功
    step5: 目标库进行数据预置; expect:执行成功
    step6: 以线程开启事务，同时进行10个dblink连接数据插入; expect: 全部插入成功
History     :
"""
import os
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

com = Common()


@unittest.skipIf(not com.check_node_exists('remote1_PrimaryDbuser'),
                 '无远程主机环境不执行')
class DblinkCase(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:初始化----')
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.remote_db = Node(node='remote1_PrimaryDbuser')
        self.remote_sh = CommonSH('remote1_PrimaryDbuser')
        self.com = Common('remote1_PrimaryDbuser')
        self.constant = Constant()
        self.db_name_1 = 'db_dblink_0023'
        self.u_sys = 'u_dblink_sys0023'

        self.tb_name = 't_dblink_case0023'
        self.pre_sql_name = 'dblink_remote_pre.sql'
        self.insert_sql_name = 'dblink_insert.sql'
        self.source_dir = os.path.join(macro.SCRIPTS_PATH, "dblink")
        self.target_pre_sql_file = os.path.join("dblink", self.pre_sql_name)
        self.target_insert_file = os.path.join("dblink", self.insert_sql_name)

        self.log.info('----远程复制sql文件----')
        com.scp_file(self.remote_db,
                     f"{self.pre_sql_name}", "dblink",
                     source_dir=self.source_dir)
        com.scp_file(self.pri_dbuser,
                     f"{self.insert_sql_name}", "dblink",
                     source_dir=self.source_dir)

    def test_main(self):
        step_txt = '----step1: 创建数据库及库下创建dblink扩展; expect:成功----'
        self.log.info(step_txt)
        db_sql = f"drop database if exists {self.db_name_1};" \
                 f"create database {self.db_name_1};"
        db_result = self.pri_sh.execut_db_sql(db_sql)
        self.log.info(db_result)
        create_sql = f"create extension dblink;"
        create_result = self.pri_sh.execut_db_sql(create_sql,
                                                  dbname=self.db_name_1)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, db_result,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.create_extension_success, create_result,
                      "执行失败" + step_txt)

        step_txt = '----step2: 目标库进行数据库及用户建立; expect:成功----'
        self.log.info(step_txt)
        db_sql = f"drop database if exists {self.db_name_1};" \
                 f"create database {self.db_name_1};"
        msg1 = self.remote_sh.execut_db_sql(db_sql,
                                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(msg1)
        user_sql = f"drop user if exists {self.u_sys} cascade;" \
                   f"create user {self.u_sys} sysadmin " \
                   f"password '{macro.PASSWD_INITIAL}';"
        msg2 = self.remote_sh.execut_db_sql(user_sql,
                                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(msg2)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, msg1,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, msg2,
                      "执行失败" + step_txt)

        step_txt = '----step3: 目标库进行pg_hba用户白名单配置; expect:成功----'
        self.log.info(step_txt)
        self.log.info('----dblink端pg_hba配置----')
        guc_res = self.remote_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  {self.db_name_1}  all '
            f'{self.pri_dbuser.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, "执行失败" + step_txt)

        step_txt = '----step4: sql脚本目标库信息变更配置; expect:成功----'
        self.log.info(step_txt)
        target_info = ['remote_ip', 'remote_db1', 'remote_port',
                       'remote_user_sys', 'remote_passwd',
                       'dblink_ip', 'remote_tb']
        reset_info = [self.remote_db.db_host, self.db_name_1,
                      self.remote_db.db_port, self.u_sys,
                      macro.PASSWD_INITIAL, self.pri_dbuser.db_host,
                      self.tb_name]
        for item_index, item in enumerate(target_info):
            msg = com.replace_file_content_by_sed(self.remote_db,
                                                  self.target_pre_sql_file,
                                                  item,
                                                  reset_info[item_index])
            self.log.info(msg)
            self.assertEqual('', msg, "执行失败" + step_txt)
            msg = com.replace_file_content_by_sed(self.pri_dbuser,
                                                  self.target_insert_file,
                                                  item,
                                                  reset_info[item_index])
            self.log.info(msg)
            self.assertEqual('', msg, "执行失败" + step_txt)

        step_txt = '----step5: 目标库进行数据预置; expect:执行成功----'
        self.log.info(step_txt)
        sql_text = self.remote_db.sh(f"cat {self.target_pre_sql_file}").result()
        self.log.info(sql_text)
        shell = f'source {macro.DB_ENV_PATH_REMOTE1};' \
                f'gsql -d {self.db_name_1} -p {self.remote_db.db_port} ' \
                f'-f {self.target_pre_sql_file}'
        self.log.info(shell)
        result = self.remote_db.sh(shell).result()
        self.log.info(result)
        self.assertNotIn('err', result.lower(), "执行失败" + step_txt)

        step6_txt = '----step6: 以线程开启事务，同时进行10个dblink连接数据插入; ' \
                    'expect: 全部插入成功----'
        self.log.info(step6_txt)
        session1 = []
        session1_result = []
        for index in range(10):
            self.pri_dbuser.sh(f'cp {self.target_insert_file} '
                               f'{self.target_insert_file}_{index}')
            msg = com.replace_file_content_by_sed(
                self.pri_dbuser, f'{self.target_insert_file}_{index}',
                'begin_index', index * 1000000)
            self.log.info(msg)
            self.assertEqual('', msg, "执行失败" + step_txt)
            msg = com.replace_file_content_by_sed(
                self.pri_dbuser, f'{self.target_insert_file}_{index}',
                'end_index', index * 1000000 + 999999)
            self.log.info(msg)
            self.assertEqual('', msg, "执行失败" + step_txt)
        for index in range(10):
            insert_cmd = f'source {macro.DB_ENV_PATH};' \
                         f'gsql -d {self.db_name_1} -p {self.pri_dbuser.db_port} ' \
                         f'-f {self.target_insert_file}_{index}'
            self.log.info(insert_cmd)
            session1.append(ComThread(self.pri_dbuser.sh,
                                      args=(insert_cmd,)))
            session1[index].setDaemon(True)
            session1[index].start()
        for index in range(10):
            session1[index].join()
            session1_result.append(session1[index].get_result().result())
        self.log.info(session1_result)
        ok_list = [item for item in session1_result if
                   'INSERT 0 1000000' in item]
        self.assertEqual(10, len(ok_list), '执行失败:' + step6_txt)

    def tearDown(self):
        self.log.info('----this is tearDown----')
        step7_txt = '----dblink侧清理数据; expect:清理成功----'
        self.log.info(step7_txt)
        drop_sql = f"drop database {self.db_name_1};" \
                   f"drop user if exists {self.u_sys} cascade;"
        msg1 = self.pri_sh.execut_db_sql(drop_sql)
        self.log.info(msg1)

        step8_txt = '----目标库进行数据库及用户删除; expect:成功----'
        self.log.info(step8_txt)
        drop_db_sql = f"clean connection to all force for database " \
                      f"{self.db_name_1};" \
                      f"drop database if exists {self.db_name_1};"
        drop_db_msg = self.remote_sh.execut_db_sql(
            drop_db_sql, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_db_msg)
        drop_user_sql = f"drop user if exists {self.u_sys} cascade;"
        drop_user_msg = self.remote_sh.execut_db_sql(
            drop_user_sql, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_user_msg)

        step9_txt = '----目标库进行pg_hba取消用户白名单配置; expect: 成功----'
        self.log.info(step9_txt)
        guc_res1 = self.remote_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  {self.db_name_1}  all '
            f'{self.pri_dbuser.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res1)

        step10_txt = '----测试脚本删除; expect: 成功----'
        self.log.info(step10_txt)
        file_rm_cmd = f'rm -rf dblink;' \
                      f'if [ -d dblink ]; ' \
                      f'then echo "exists"; else echo "not exists"; fi'
        self.log.info(file_rm_cmd)
        file_rm_result1 = self.remote_db.sh(file_rm_cmd).result()
        self.log.info(file_rm_result1)
        file_rm_result2 = self.pri_dbuser.sh(file_rm_cmd).result()
        self.log.info(file_rm_result2)

        self.log.info(f'----{os.path.basename(__file__)}:执行完毕----')
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, msg1,
                      "执行失败" + step7_txt)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, msg1,
                      "执行失败" + step7_txt)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, drop_db_msg,
                      "执行失败" + step8_txt)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_user_msg,
                      "执行失败" + step8_txt)
        self.assertTrue(guc_res1, "执行失败" + step9_txt)
        self.assertEqual('not exists', file_rm_result1, "执行失败" + step10_txt)
        self.assertEqual('not exists', file_rm_result2, "执行失败" + step10_txt)
